package com.exd.flink.stream.function;

import java.util.Iterator;
import java.util.Set;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import com.exd.flink.common.DomainControlFlag;
import com.exd.flink.common.GzipUtils;
import com.exd.flink.common.OutputVComputeTagBase;
import com.exd.flink.common.StringTool;
import com.exd.flink.domain.AlgorithmStatusBean;
import com.exd.flink.domain.ResourcesBean;
import com.exd.flink.domain.VComputeBean;
import com.exd.flink.domain.VersionBean;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;

/**
 * 解析转换VCOMPUTE的数据
 */
public class VcomputeAnalyticalFunction extends ProcessFunction<String, VComputeBean> {

	private static final long serialVersionUID = 1L;

	@Override
	public void processElement(String message, Context context, Collector<VComputeBean> collector) throws Exception {
		try {
			JsonElement msgElement = JsonParser.parseString(message);
			JsonObject jsonObject = msgElement.getAsJsonObject();
			// 封装公共数据
			VComputeBean vComputeBean = new VComputeBean();
			vComputeBean.setVin(jsonObject.get("vin").getAsString());
			vComputeBean.setServiceId(jsonObject.get("serviceId").getAsString());
			vComputeBean.setMethodId(jsonObject.get("methodId").getAsLong());
			vComputeBean.setMajorVersion(jsonObject.get("majorVersion").getAsLong());
			vComputeBean.setSamplingTime(jsonObject.get("samplingTime").getAsLong());
			vComputeBean.setTboxSendTimestamp(jsonObject.get("tboxSendTimestamp").getAsLong());
			vComputeBean.setReceiveTimestamp(jsonObject.get("receiveTimestamp").getAsLong());
			// 获取data数据
			String data = jsonObject.get("data").getAsString();
			String decodedDataString = GzipUtils.uncompress(data);
			JsonArray asJsonArray = JsonParser.parseString(decodedDataString).getAsJsonArray();
			Iterator<JsonElement> iterator = asJsonArray.iterator();
			while(iterator.hasNext()) {
				JsonObject asJsonObject = iterator.next().getAsJsonObject();
				// vcompute数据处理
				JsonElement dFlagData = asJsonObject.get("d");
				if(dFlagData!=null) assemblyVcomputeData(vComputeBean,asJsonObject,dFlagData,context);
				// 资源监控数据
				JsonElement rsFlagData = asJsonObject.get("rs");
				if(rsFlagData!=null)assemblyResourcesData(vComputeBean,rsFlagData,context);
			}	
		} catch (Exception e) {
			System.err.println(e.getMessage());
		}
	}
	
	/**
	 * 描述: 处理vcompute数据
	 * 创建人: 赵兴炎
	 * 日期: 2022年9月13日
	 */
	public void assemblyVcomputeData(VComputeBean vComputeBean,JsonObject asJsonObject,JsonElement dFlagData,Context context) {
		String header = asJsonObject.get("p").getAsJsonArray().toString();
		vComputeBean.setHeader(header);
		long startTime=dFlagData.getAsJsonArray().get(0).getAsJsonArray().get(0).getAsLong();
		vComputeBean.setCollectTime(startTime);
		JsonObject rolloutObj = asJsonObject.getAsJsonObject("vrollout");
		Long deployId = rolloutObj.get("dsId").getAsLong();
		vComputeBean.setDeployId(deployId);
		// 输出正常的数据,根据不同的域控，输出到不同的流
		if (vComputeBean.getServiceId().equalsIgnoreCase(DomainControlFlag.BDCM_VCOMPUTE)) {
			context.output(OutputVComputeTagBase.BDCM_ANALYTICAL_VCOMPUTE_OK, vComputeBean);
		} else if (vComputeBean.getServiceId().equalsIgnoreCase(DomainControlFlag.VDCM_VCOMPUTE)) {
			context.output(OutputVComputeTagBase.VDCM_ANALYTICAL_VCOMPUTE_OK, vComputeBean);
		}
	}
	
	/**
	 * 描述: 处理资源监控数据
	 * 创建人: 赵兴炎
	 * 日期: 2022年9月13日
	 */
	public void assemblyResourcesData(VComputeBean vComputeBean,JsonElement rsFlagData,Context context) {
		JsonArray asJsonArray = rsFlagData.getAsJsonArray();
		Iterator<JsonElement> iterator = asJsonArray.iterator();
		while(iterator.hasNext()) {
			JsonObject asJsonObject = iterator.next().getAsJsonObject();
			// 版本数据
			JsonElement versionData = asJsonObject.get("v");
			if(versionData!=null) {
				VersionBean versionBean = new VersionBean();
				versionBean.setVin(vComputeBean.getVin());
				versionBean.setServiceId(vComputeBean.getServiceId());
				JsonObject versionObject = versionData.getAsJsonObject();
				versionBean.setLibrary(versionObject.get("library").getAsString());
				versionBean.setVcompute(versionObject.get("vcompute").getAsString());
				versionBean.setVdata(versionObject.get("vdata").getAsString());
				JsonElement matrixData = versionObject.get("matrix");
				if(matrixData!=null) {
					versionBean.setMatrix(matrixData.getAsString());
				}
				JsonElement phraseData = versionObject.get("phrase");
				if(phraseData!=null) {
					versionBean.setPhrase(phraseData.getAsString());
				}
				JsonElement vehicleData = versionObject.get("vehicle");
				if(vehicleData!=null) {
					versionBean.setVehicle(vehicleData.getAsString());
				}
				context.output(OutputVComputeTagBase.VERSION_DATA, versionBean);
			}
			// BDCM资源监控数据
			JsonElement vComputeResData = asJsonObject.get("vCompute");
			JsonElement vDataResData = asJsonObject.get("vData");
			if(vComputeResData!=null || vDataResData!=null) {
				assemblyBdcmResourcesData(vComputeBean,vComputeResData,vDataResData,context);
			}
			// VDCM资源监控数据
			JsonElement cpumemoryData = asJsonObject.get("s");
			if(cpumemoryData!=null)assemblyVdcmResourcesData(vComputeBean,cpumemoryData,context);
			// 获取算法状态数据
			if(asJsonObject.toString().contains("status")) {
				Set<String> keySet = asJsonObject.keySet();
				keySet.forEach(dataKey -> {
					JsonElement jsonElement = asJsonObject.get(dataKey);
					JsonObject statusObject = jsonElement.getAsJsonObject();
					String vin = statusObject.get("deviceId").getAsString();
					String deployId = statusObject.get("dsId").getAsString();
					int status = statusObject.get("status").getAsInt();
					AlgorithmStatusBean algorithmStatusBean = new AlgorithmStatusBean();
					algorithmStatusBean.setAlgorithmId(dataKey);
					algorithmStatusBean.setVin(vin);
					algorithmStatusBean.setDeployId(deployId);
					algorithmStatusBean.setStatus(status);
					context.output(OutputVComputeTagBase.ALGORITHM_STATUS_DATA, algorithmStatusBean);
				});
			}
		}
	}
	
	/**
	 * 描述: 处理bdcm资源监控数据
	 * 创建人: 赵兴炎
	 * 日期: 2022年9月13日
	 */
	public void assemblyBdcmResourcesData(VComputeBean vComputeBean,JsonElement vComputeResData,JsonElement vDataResData,Context context) {
		ResourcesBean resourcesBean = new ResourcesBean();
		resourcesBean.setVin(vComputeBean.getVin());
		resourcesBean.setServiceId(vComputeBean.getServiceId());
		resourcesBean.setServiceName("bdcm");
		resourcesBean.setSamplingTime(vComputeBean.getSamplingTime());
		// vcompute的内存和CPU数据
		if(vComputeResData!=null) {
			String[] tageValues = vComputeResData.getAsString().trim().split("\\s+");
			resourcesBean.setVComputeCpu(Math.round(Float.parseFloat(tageValues[8]) / 5));
			resourcesBean.setVComputeMem(Integer.parseInt(tageValues[5]));
		}
		// vdata的内存和CPU数据
		if(vDataResData!=null) {
			String[] tageValues = vDataResData.getAsString().trim().split("\\s+");
			resourcesBean.setVDataCpu(Math.round(Float.parseFloat(tageValues[8]) / 5));
			resourcesBean.setVDataMem(Integer.parseInt(tageValues[5]));
		}
		context.output(OutputVComputeTagBase.RESOUCRES_DATA, resourcesBean);
	}
	
	/**
	 * 描述: 处理vdcm资源监控数据
	 * 创建人: 赵兴炎
	 * 日期: 2022年9月13日
	 */
	public void assemblyVdcmResourcesData(VComputeBean vComputeBean,JsonElement rsFlagData,Context context) {
		JsonObject resourObject = rsFlagData.getAsJsonObject();
		JsonElement vdcmComputeResData = resourObject.get("vCompute");
		JsonElement vdcmDataResData = resourObject.get("vData");
		if(vdcmComputeResData!=null || vdcmDataResData!=null) {
			ResourcesBean resourcesBean = new ResourcesBean();
			resourcesBean.setVin(vComputeBean.getVin());
			resourcesBean.setServiceId(vComputeBean.getServiceId());
			resourcesBean.setServiceName("vdcm");
			resourcesBean.setSamplingTime(vComputeBean.getSamplingTime());
			// vcompute的内存和CPU数据
			if(vdcmComputeResData!=null) {
				String[] tageValues = vdcmComputeResData.getAsString().trim().split("\\s+");
				resourcesBean.setVComputeCpu(StringTool.resourceValueStringToInt(tageValues[3], tageValues[3].length()-1));
				resourcesBean.setVComputeMem(StringTool.resourceValueStringToInt(tageValues[5], tageValues[5].length()-1));
			}
			// vdata的内存和CPU数据
			if(vdcmDataResData!=null) {
				String[] tageValues = vdcmDataResData.getAsString().trim().split("\\s+");
				resourcesBean.setVDataCpu(StringTool.resourceValueStringToInt(tageValues[3], tageValues[3].length()-1));
				resourcesBean.setVDataMem(StringTool.resourceValueStringToInt(tageValues[5], tageValues[5].length()-1));
			}
			context.output(OutputVComputeTagBase.RESOUCRES_DATA, resourcesBean);
		}
	}

}
